package com.gandong8.hbase.tool.domain;

import com.gandong8.hbase.tool.annotation.HbaseColumn;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.Date;
@Data
@Slf4j
public class HbaseQuery {
    String rowKey;
    String rowPrefix;
    String startRow;
    String stopRow;
    Long minStamp;
    Long maxStamp;
    Long timeStamp;
    Integer pageSize;
    String qualifier;
    final Scan scan = new Scan();

    public HbaseQuery() {
    }

    public HbaseQuery(Object obj) {
        if(obj != null) {
            columnFilter(obj, CompareFilter.CompareOp.EQUAL);
        }
    }

    public HbaseQuery(Object obj, CompareFilter.CompareOp compareOp) {
        if(obj != null) {
            columnFilter(obj, compareOp);
        }
    }

    public void columnFilter(Object obj, CompareFilter.CompareOp compareOp) {
        Field[] fields = obj.getClass().getDeclaredFields();
        for (Field field : fields) {
            HbaseColumn column = field.getAnnotation(HbaseColumn.class);
            if (column == null) {
                continue;
            }
            field.setAccessible(true);
            String family = column.family();
            String qualifier = column.qualifier();
            Object fieldObject = null;
            try {
                fieldObject = field.get(obj);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
            if (fieldObject == null) {
                continue;
            }
            String value;
            if (fieldObject instanceof Date) {
                value = String.valueOf(((Date) fieldObject).getTime());
            } else {
                value = fieldObject.toString();
            }
            Filter filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier), compareOp, Bytes.toBytes(value));
            scan.setFilter(filter);
            break;
        }
    }

    public Scan getScan() {
        if(rowPrefix != null) {
            scan.setRowPrefixFilter(Bytes.toBytes(rowPrefix));
        }
        if(startRow != null) {
            scan.setStartRow(Bytes.toBytes(startRow));
        }
        if(stopRow != null) {
            scan.setStopRow(Bytes.toBytes(stopRow));
        }
        if(maxStamp != null && minStamp !=null) {
            try {
                //condition: >=minStamp and <maxStamp
                scan.setTimeRange(minStamp, maxStamp);
            } catch (IOException e) {
                log.error("setTimeRange_err", e);
            }
        }
        if(timeStamp != null) {
            try {
                scan.setTimeStamp(timeStamp);
            } catch (IOException e) {
                log.error("setTimeStamp_err", e);
            }
        }
        if(pageSize != null) {
            //scan.setLimit(100)
            scan.setFilter(new PageFilter(pageSize));
        }
        scan.setReversed(true);
        scan.setMaxResultSize(2 * 1024 * 1024);
        return scan;
    }

}
